package com.sergio.compress.model.channelWapper;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.avro.AvroFactory;
import com.fasterxml.jackson.dataformat.avro.AvroMapper;
import com.fasterxml.jackson.dataformat.avro.AvroSchema;
import com.fasterxml.jackson.dataformat.avro.schema.AvroSchemaGenerator;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.sergio.compress.model.base.ChannelWrapper;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.StopWatch;

import org.apache.avro.Schema;

import java.util.List;

/**
 * @Author : Sergio.D
 * @Description :
 * @CreateDate ：2023/6/15 20:02
 */
@Data
public abstract class JacksonAvroWrapper<T> extends ChannelWrapper<T> {

    private static String SCHEMA;

    @Override

    public String getChannelName() {
        return "Jackson-Avro";
    }


    @Override
    public byte[] serialize() {
        try {
            getSchema();
            AvroMapper mapper = new AvroMapper();
            mapper.registerModule(new JavaTimeModule());
            Schema raw = new Schema.Parser().setValidate(true).parse(SCHEMA);
            AvroSchema schema = new AvroSchema(raw);

            StopWatch stopWatch = new StopWatch();
            stopWatch.start();
            byte[] avroData = mapper.writer(schema)
                    .writeValueAsBytes(this.dataSend);
            stopWatch.stop();
            timeSe = stopWatch.getTotalTimeNanos();
            lengthSource = avroData.length;
            return avroData;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void deserialize(byte[] source) {
        try {// 创建ProtobufMapper实例

            AvroMapper mapper = new AvroMapper();
            mapper.registerModule(new JavaTimeModule());
            Schema raw = new Schema.Parser().setValidate(true).parse(SCHEMA);
            AvroSchema schema = new AvroSchema(raw);

            StopWatch stopWatch = new StopWatch();
            stopWatch.start();

            T otherValue = mapper.reader(getMainClazz())
                    .with(schema)
                    .readValue(source);

//            List<TestDataBase> otherValue = mapper.readValue(source, mapper.getTypeFactory().constructCollectionType(List.class, TestDataBase.class));
            stopWatch.stop();
            timeDse = stopWatch.getTotalTimeNanos();
            this.resultSe = this.dataSend.equals(otherValue);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        if (!this.resultSe) {
            throw new RuntimeException("反解析结果不一致!!!");
        }
    }

    public void getSchema() {
        if (StringUtils.isBlank(SCHEMA)) {
            try {
                ObjectMapper mapper = new ObjectMapper(new AvroFactory());
                mapper.registerModule(new JavaTimeModule());
                AvroSchemaGenerator gen = new AvroSchemaGenerator();
                mapper.acceptJsonFormatVisitor(getMainClazz(), gen);
                AvroSchema schemaWrapper = gen.getGeneratedSchema();

                org.apache.avro.Schema avroSchema = schemaWrapper.getAvroSchema();
                String asJson = avroSchema.toString(true);
                // 获取Avro schema的原始字符串表示
                SCHEMA = schemaWrapper.getAvroSchema().toString();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

    }

}
